package com.sxh.logging;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Clob;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
//参考AI
@Component
public class RocketMQConsumerUtil implements InitializingBean {

    @Autowired
    private LogMap logMap;

    private final String NAME_SERVER_ADDR = "127.0.0.1:9876";
    private final String CONSUMER_GROUP = "szml";
    private final String TOPIC = "UserOption";

    @Override
    public void afterPropertiesSet() throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        // 设置NameServer地址
        consumer.setNamesrvAddr(NAME_SERVER_ADDR);

        // 订阅主题和标签（这里订阅所有标签的消息）
        consumer.subscribe(TOPIC, "*");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context
            ) {
                for (MessageExt msg : msgs) {
                    try {
                        // 处理接收到的消息
                        processMessage(msg);
                    } catch (Exception e) {
                        System.err.println("消息处理失败: " + e.getMessage());
                        // 可以返回RECONSUME_LATER进行重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("消费者已启动");
    }

    public void processMessage(MessageExt msg) {
        try {
            // 解析消息体
            String jsonStr = new String(msg.getBody(), "UTF-8");
            System.out.println("接收到消息: " + jsonStr);
            // 解析JSON并插入数据库（示例逻辑）
            insertLogToDatabase(jsonStr);
        } catch (Exception e) {
            System.err.println("消息解析失败: " + e.getMessage());
        }
    }
    @Transactional
    public void insertLogToDatabase(String jsonStr) {
        // 解析消息
        Map<String, Object> map = JSON.parseObject(jsonStr);
        OperationLog log = new OperationLog(
                null,
                Long.parseLong(map.get("userId").toString()),
                map.get("action").toString(),
                map.get("ip").toString(),
                map.get("detail").toString(),
                Long.parseLong(map.get("optionId").toString()),
                LocalDateTime.parse(map.get("optionTime").toString())
        );
        // 通过注入的logMap实例调用insertLog方法
        logMap.insertLog(log);
        System.out.println("插入数据库: " + jsonStr);
    }
}